package com.ustc.hewei.nettyrpc2_0.netty.client.manager;

import com.ustc.hewei.nettyrpc2_0.common.config.RpcProperties;
import com.ustc.hewei.nettyrpc2_0.common.factory.ThreadPoolFactory;
import com.ustc.hewei.nettyrpc2_0.common.utils.Constants;
import com.ustc.hewei.nettyrpc2_0.common.utils.PropertiesUtil;
import com.ustc.hewei.nettyrpc2_0.common.warp.ZkNode;
import com.ustc.hewei.nettyrpc2_0.netty.client.handler.ClientHandler;
import com.ustc.hewei.nettyrpc2_0.netty.client.handler.ClientInitializerHandler;
import com.ustc.hewei.nettyrpc2_0.netty.client.loadBalance.CommonLoadBalance;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import javafx.util.Pair;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author hewei
 * @version 1.0
 * @description: 客户端每个连接的handler管理类
 * @date 2022/9/6 10:49
 */

@Slf4j
public class HandlerManager {
    // TODO: 2022/9/19 存在自带的provider，可以使用进行简化
    private final Map<InetSocketAddress, ClientHandler> handlerCache;
    private final Map<String, List<ZkNode>> serviceCache;
    private final Map<InetSocketAddress, List<ZkNode>> addressCache;
    // 存放降级的服务，使用队列，按时间顺序排列
    private final ConcurrentLinkedDeque<Pair<ZkNode, Long>> invalidServices;
    private final Bootstrap bootstrap;
    private final ExecutorService connectProvideThreadPool;
    private final CommonLoadBalance loadBalance;
    private final ReentrantLock lock;
    private final Condition connected;
    private final Long ZK_TIMEOUT;

    public HandlerManager(CommonLoadBalance loadBalance) {
        handlerCache = new ConcurrentHashMap<>();
        serviceCache = new ConcurrentHashMap<>();
        addressCache = new ConcurrentHashMap<>();
        lock = new ReentrantLock();
        connected = lock.newCondition();
        bootstrap = new Bootstrap().channel(NioSocketChannel.class)
                .group(new NioEventLoopGroup())
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true);
        connectProvideThreadPool = ThreadPoolFactory.createDefaultThreadPool("connectProvidePool");
        this.loadBalance = loadBalance;
        invalidServices = new ConcurrentLinkedDeque<>();
        ZK_TIMEOUT = 20000L;
        //ScheduledExecutorService invalidHandlerThreadPool = Executors.newScheduledThreadPool(1);
        ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
        scheduledThreadPoolExecutor.setMaximumPoolSize(1);
        scheduledThreadPoolExecutor.scheduleAtFixedRate(this::detectInvalidService, 1, ZK_TIMEOUT, TimeUnit.MILLISECONDS);
        //invalidHandlerThreadPool.scheduleAtFixedRate(this::detectInvalidService, 1, ZK_TIMEOUT, TimeUnit.MILLISECONDS);
    }

    public void connectToProvide(ZkNode zkNode) {
        InetSocketAddress address = new InetSocketAddress(zkNode.getHost(), zkNode.getPort());
        if (!serviceCache.containsKey(zkNode.getServiceName())) {
            serviceCache.put(zkNode.getServiceName(), new ArrayList<>());
        }
        if (!addressCache.containsKey(address)) {
            // 使用LinkedList是因为加快remove的速度
            addressCache.put(address, new LinkedList<>());
        }
        if (handlerCache.containsKey(address)) {
            serviceCache.get(zkNode.getServiceName()).add(zkNode);
            addressCache.get(address).add(zkNode);
        } else {
            connectProvideThreadPool.submit(() -> {
                ChannelFuture channelFuture = bootstrap.handler(new ClientInitializerHandler()).connect(address);
                channelFuture.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture channelFuture) {
                        if (channelFuture.isSuccess()) {
                            ClientHandler handler = channelFuture.channel().pipeline().get(ClientHandler.class);
                            log.info("客户端已连接到服务 {}", zkNode);
                            // 防止一个地址提供多个服务，启动瞬间，因为消费者监听器是异步的，
                            // 所以会同时进行两次更新，导致两次连接到相同的节点，此时应将新的连接直接关闭
                            if (!handlerCache.containsKey(address)) {
                                handlerCache.put(address, handler);
                            } else {
                                handler.close();
                            }
                            addressCache.get(address).add(zkNode);
                            serviceCache.get(zkNode.getServiceName()).add(zkNode);
                            signalAvailableHandler();
                        }
                    }
                });
            });
        }
    }

    private void signalAvailableHandler() {
        lock.lock();
        try {
            connected.signalAll();
        } finally {
            lock.unlock();
        }
    }

    public ClientHandler chooseHandler(String serviceName) throws Exception {
        List<ZkNode> zkNodes = serviceCache.get(serviceName);
        if (zkNodes == null || zkNodes.size() <= 0) {
            waitForHandler(serviceName);
            zkNodes = serviceCache.get(serviceName);
        }
        if (zkNodes == null || zkNodes.size() <= 0) {
            log.error("未找到可用的服务提供者 {}", serviceName);
            return null;
        }
        ZkNode zkNode = loadBalance.getService(zkNodes);
        if (zkNode == null) {
            return null;
        }
        ClientHandler handler = handlerCache.get(new InetSocketAddress(zkNode.getHost(), zkNode.getPort()));
        if (handler != null) {
            return handler;
        }
        log.error("未找到可用的服务提供者 {}", serviceName);
        return null;
    }

    public void waitForHandler(String serviceName) throws InterruptedException {
        lock.lock();
        try {
            log.info("等待服务的handler {} 连接成功...", serviceName);
            connected.await(PropertiesUtil.getLong(Constants.waitTimeout), TimeUnit.MILLISECONDS);
        } finally {
            lock.unlock();
        }
    }

    // 更新服务的步骤分为，对于新的地址需要建立连接，对于目前不存在的地址需要删除地址下的所有服务并关闭连接，对于老的地址重新写入到缓存中
    public void updateCache(String serviceName, List<ZkNode> list) {
        List<ZkNode> oldCache = serviceCache.get(serviceName);
        HashSet<ZkNode> nodeHashSet;
        if (oldCache == null) {
            nodeHashSet = new HashSet<>();
        } else {
            nodeHashSet = new HashSet<>(oldCache);
        }
        List<ZkNode> waitForReConnect = new ArrayList<>();
        List<ZkNode> newList = new ArrayList<>();
        for (ZkNode zkNode : list) {
            if (!nodeHashSet.remove(zkNode)) {
                waitForReConnect.add(zkNode);
            } else {
                newList.add(zkNode);
                zkNode.setCur(0);
            }
        }
        // 保存不变的节点
        serviceCache.put(serviceName, newList);
        // 对淘汰的节点进行处理
        for (ZkNode zkNode : nodeHashSet) {
            InetSocketAddress address = new InetSocketAddress(zkNode.getHost(), zkNode.getPort());
            ClientHandler handler = handlerCache.get(address);
            List<ZkNode> nodes = addressCache.get(address);
            if (handler != null && nodes != null) {
                // 将地址下的对应服务删除
                nodes.remove(zkNode);
                // 如果该地址下已经没有服务了，那么关闭连接
                if (nodes.isEmpty()) {
                    handler.close();
                    handlerCache.remove(address);
                    addressCache.remove(address);
                    log.info("地址 {} 下已无有效服务!", address);
                }
            }
        }
        // 对新节点进行处理
        for (ZkNode zkNode : waitForReConnect) {
            connectToProvide(zkNode);
        }
    }

    public ClientHandler retried(ClientHandler handler, String serviceName) throws Exception {
        InetSocketAddress address = handler.getRemoteAddress();
        ZkNode tempNode = new ZkNode();
        tempNode.setHost(address.getHostName()).setPort(address.getPort()).setServiceName(serviceName);
        // 加到失效服务中
        // TODO: 2022/10/26 在多线程下，会存在多次添加的问题
        invalidServices.addLast(new Pair<>(tempNode, System.currentTimeMillis()));
        List<ZkNode> zkNodes = addressCache.get(address);
        if (zkNodes != null) {
            for (ZkNode zkNode : zkNodes) {
                // 只对地址下的对应服务降级，而不是对整个地址下的所有服务降级
                if (serviceName.equals(zkNode.getServiceName())) zkNode.setWeight(0);
            }
        }
        log.info("重试获取handler...");
        return chooseHandler(serviceName);
    }

    public void detectInvalidService() {
        long currentTimeMillis = System.currentTimeMillis();
        while (!invalidServices.isEmpty()) {
            Pair<ZkNode, Long> invalid = invalidServices.pollFirst();
            Long lastTimestamp = invalid.getValue();
            ZkNode tempNode = invalid.getKey();
            if (currentTimeMillis - lastTimestamp <= ZK_TIMEOUT) {
                // 当前队列头时间间隔小于设定时间，将node放回队列头，并直接结束
                invalidServices.addFirst(invalid);
                break;
            }
            recoverService(new InetSocketAddress(tempNode.getHost(), tempNode.getPort()), tempNode.getServiceName());
        }
    }

    public void recoverService(InetSocketAddress address, String serviceName) {
        int weight = PropertiesUtil.getInt(Constants.defaultWeight);
        List<ZkNode> zkNodes = addressCache.get(address);
        if (zkNodes != null) {
            for (ZkNode zkNode : zkNodes) {
                // 只对降级的服务设置初始权重
                if (zkNode.getServiceName().equals(serviceName)) zkNode.setWeight(weight);
            }
            log.info("节点 {} 恢复成功!", address);
        } else {
            log.info("节点 {} 已经清除!", address);
        }
    }
}